package com.airlenet.dubbo.registry.hazelcast;

import com.hazelcast.core.*;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.support.FailbackRegistry;
import org.apache.dubbo.rpc.RpcException;

import java.io.Serializable;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static java.util.concurrent.Executors.newCachedThreadPool;
import static org.apache.dubbo.common.constants.CommonConstants.*;
import static org.apache.dubbo.common.constants.RegistryConstants.CATEGORY_KEY;
import static org.apache.dubbo.common.constants.RegistryConstants.DEFAULT_CATEGORY;
import static org.apache.dubbo.registry.Constants.*;

public class HazelcastRegistry extends FailbackRegistry implements MembershipListener {

    private static final Logger logger = LoggerFactory.getLogger(HazelcastRegistry.class);

    private final int expirePeriod;
    private ExecutorService notifierExecutor = newCachedThreadPool(
            new NamedThreadFactory("dubbo-hazelcast-notifier", true));
    private final ConcurrentMap<URL, Notifier> notifiers = new ConcurrentHashMap<>();
    private final static String DEFAULT_ROOT = "dubbo";
    HazelcastInstance hazelcastInstance;
    private final String root;

    public HazelcastRegistry(URL url, HazelcastInstance hazelcastInstance) {
        super(url);
        if (url.isAnyHost()) {
            throw new IllegalStateException("hazelcast address == null");
        }
        String group = url.getParameter(GROUP_KEY, DEFAULT_ROOT);
        if (!group.startsWith(PATH_SEPARATOR)) {
            group = PATH_SEPARATOR + group;
        }
        if (!group.endsWith(PATH_SEPARATOR)) {
            group = group + PATH_SEPARATOR;
        }
        this.root = group;

        this.expirePeriod = url.getParameter(SESSION_TIMEOUT_KEY, DEFAULT_SESSION_TIMEOUT);
        this.hazelcastInstance = hazelcastInstance;
        this.hazelcastInstance.getCluster().addMembershipListener(this);

    }

    @Override
    public void register(URL url) {
        if (isConsumerSide(url)) {
            return;
        }

        super.register(url);
    }

    @Override
    public void doRegister(URL url) {
        String service = url.getServiceInterface();

        hazelcastInstance.getMultiMap("dubbo-service").put(service, url);
        String uuid = hazelcastInstance.getCluster().getLocalMember().getUuid();
        hazelcastInstance.getMultiMap("dubbo-provider").put(uuid, new HazelcastService(service, url));
    }

    @Override
    public void unregister(URL url) {
        if (isConsumerSide(url)) {
            return;
        }

        super.unregister(url);
    }

    @Override
    public void doUnregister(URL url) {
        String service = url.getServiceInterface();

        hazelcastInstance.getMultiMap("dubbo-service").remove(service, url);
        String uuid = hazelcastInstance.getCluster().getLocalMember().getUuid();
        Iterator<Object> iterator = hazelcastInstance.getMultiMap("dubbo-provider").get(uuid).iterator();
        while (iterator.hasNext()) {
            HazelcastService next = (HazelcastService) iterator.next();
            if (next.getService().equals(service)) {
                iterator.remove();
                break;
            }
        }
    }

    @Override
    public void subscribe(URL url, NotifyListener listener) {
        if (isProviderSide(url)) {
            return;
        }

        super.subscribe(url, listener);
    }

    @Override
    public void doSubscribe(URL url, NotifyListener listener) {


        List<URL> urls;
        if (ANY_VALUE.equals(url.getServiceInterface())) {
            urls = hazelcastInstance.getMultiMap("dubbo-service").values()
                    .stream().map(v -> (URL) (v)).collect(Collectors.toList());

        } else {
            String service = url.getServiceInterface();
            urls = hazelcastInstance.getMultiMap("dubbo-service").get(service)
                    .stream().map(v -> (URL) (v)).collect(Collectors.toList());
        }

        notify(url, listener, urls);
        Notifier notifier = notifiers.computeIfAbsent(url, k -> new Notifier(url));
        notifierExecutor.submit(notifier);

    }

    @Override
    public void unsubscribe(URL url, NotifyListener listener) {
        if (isProviderSide(url)) {
            return;
        }

        super.unsubscribe(url, listener);
    }

    @Override
    public void doUnsubscribe(URL url, NotifyListener listener) {
        Notifier notifier = notifiers.remove(url);
        notifier.stop();
    }

    @Override
    public List<URL> lookup(URL url) {
        if (url == null) {
            throw new IllegalArgumentException("lookup url == null");
        }
        try {
            String service = url.getServiceKey();
            return hazelcastInstance.getMultiMap("dubbo-service").get(service)
                    .stream().map(v -> (URL) (v)).collect(Collectors.toList());
        } catch (Throwable e) {
            throw new RpcException("Failed to lookup " + url + " from consul " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

    @Override
    public boolean isAvailable() {
        return hazelcastInstance != null;
    }

    @Override
    public void destroy() {
        super.destroy();
        hazelcastInstance.shutdown();
        hazelcastInstance = null;
    }

    private boolean isConsumerSide(URL url) {
        return url.getProtocol().equals(CONSUMER_PROTOCOL);
    }

    private boolean isProviderSide(URL url) {
        return url.getProtocol().equals(PROVIDER_PROTOCOL);
    }

    @Override
    public void memberAdded(MembershipEvent membershipEvent) {

    }

    @Override
    public void memberRemoved(MembershipEvent membershipEvent) {
        String uuid = membershipEvent.getMember().getUuid();

        MultiMap<String, HazelcastService> list = hazelcastInstance.getMultiMap("dubbo-provider");
        Collection<HazelcastService> hazelcastServices = list.remove(uuid);

        hazelcastServices.forEach(hazelcastService -> {
            hazelcastInstance.getMultiMap("dubbo-service").remove(hazelcastService.getService(), hazelcastService.getUrl());
        });
    }

    @Override
    public void memberAttributeChanged(MemberAttributeEvent memberAttributeEvent) {

    }


    private class Notifier implements Runnable {
        private URL url;
        private boolean running;

        Notifier(URL url) {
            this.url = url;
            this.running = true;
        }

        @Override
        public void run() {
            while (this.running) {
                if (ANY_VALUE.equals(url.getServiceInterface())) {
                    processServices();
                } else {
                    processService();
                }
                try {
                    Thread.sleep(expirePeriod);
                } catch (InterruptedException e) {
                    // ignore
                }
            }
        }

        private void processService() {
            String service = url.getServiceKey();

            List<URL> urls = hazelcastInstance.getMultiMap("dubbo-service").get(service).stream().map(v -> (URL) (v)).collect(Collectors.toList());
            for (NotifyListener listener : getSubscribed().get(url)) {
                doNotify(url, listener, urls);
            }
        }

        private void processServices() {
            List<URL> urls = hazelcastInstance.getMultiMap("dubbo-service").values().stream().map(v -> (URL) (v)).collect(Collectors.toList());
            for (NotifyListener listener : getSubscribed().get(url)) {
                doNotify(url, listener, urls);
            }
        }

        void stop() {
            this.running = false;
        }
    }

    private static class HazelcastService implements Serializable {
        private String service;
        private URL url;

        public HazelcastService(String service, URL url) {
            this.service = service;
            this.url = url;
        }

        public String getService() {
            return service;
        }

        public URL getUrl() {
            return url;
        }

        @Override
        public String toString() {
            return "HazelcastService{" +
                    "service='" + service + '\'' +
                    ", url=" + url +
                    '}';
        }
    }
}
